package com.atguigu.chapter05.sink;

import com.alibaba.fastjson.JSON;
import com.atguigu.bean.WaterSensor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.redis.RedisSink;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommand;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommandDescription;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper;

import java.util.ArrayList;

/**
 * Author: Pepsi
 * Date: 2023/7/29
 * Desc:
 */
public class Flink03_Redis_1 {
    public static void main(String[] args) {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port",1000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);

        ArrayList<WaterSensor> waterSensors = new ArrayList<>();
        waterSensors.add(new WaterSensor("sensor_1", 1607527992000L, 20));
        waterSensors.add(new WaterSensor("sensor_1", 1607527994000L, 50));
        waterSensors.add(new WaterSensor("sensor_1", 1607527996000L, 50));
        waterSensors.add(new WaterSensor("sensor_2", 1607527993000L, 10));
        waterSensors.add(new WaterSensor("sensor_2", 1607527995000L, 30));

        DataStreamSource<WaterSensor> stream = env.fromCollection(waterSensors);

        SingleOutputStreamOperator<WaterSensor> result = stream
                .keyBy(WaterSensor::getId)
                .sum("vc");

        FlinkJedisPoolConfig config = new FlinkJedisPoolConfig.Builder()
                .setHost("hadoop101")
                .setPort(6379)
                .setMaxTotal(100)  //最大连接数量
                .setMaxIdle(10)  // 连接池里面的最大空闲
                .setMinIdle(2)   // 连接池里面的最小空闲
                .setTimeout(10*1000)  // 超时时间
                .build();
        // 写出到redis中
        result.addSink(new RedisSink<>(config, new RedisMapper<WaterSensor>() {
            // 返回命令描述符：往不同的数据结构写数据用的方法不一样
            @Override
            public RedisCommandDescription getCommandDescription() {
                // 写入到字符串，用set
                return new RedisCommandDescription(RedisCommand.SET);
            }

            @Override
            public String getKeyFromData(WaterSensor waterSensor) {
                return waterSensor.getId();
            }

            @Override
            public String getValueFromData(WaterSensor waterSensor) {
                return JSON.toJSONString(waterSensor);
            }
        }));

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
/*
往redis里面写字符串，string   命令提示符用set
假设写的key是id,value是整个json格式的字符串
key         value
sensor_1    json格式字符串
 */